package com.we.risk.behavior.cross;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.we.utils.CommonTimeTools;
import java.math.BigDecimal;
import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static com.we.risk.behavior.cross.CrossCommonVars.DATERANGE;

public class TableFdmHfqLoanCenterRiskRecord {
    public static final String KFKTABLE = "cdh6_hive.fdm.hhl01_hfq_loancenter_risk_record_230620";

    public static final String QUERYSQL = "select user_key, loan_id, loan_line, " +
            "avail_line as loan_avail_line, loan_rate, " +
            "`period` as apply_period, amount as apply_amount, " +
            "create_time as apply_time, avail_periods as loan_period, " +
            "uid from " + KFKTABLE +
            " where user_key is not null " +
            "and create_time is not null " +
            "and avail_periods is not null";

    public String user_key;

    public Long loan_id;

    public BigDecimal loan_line;

    public BigDecimal loan_avail_line;

    public BigDecimal loan_rate;

    public int apply_period;

    public BigDecimal apply_amount;

    public Timestamp apply_time;

    public String loan_period;

    public Long uid;

    public static Logger LOG = LoggerFactory.getLogger(TableFdmHfqLoanCenterRiskRecord.class);

    public TableFdmHfqLoanCenterRiskRecord() {}

    public TableFdmHfqLoanCenterRiskRecord(String user_key, Long loan_id, BigDecimal loan_line,
                                           BigDecimal loan_avail_line, BigDecimal loan_rate, int apply_period,
                                           BigDecimal apply_amount, Timestamp apply_time, String loan_period, Long uid) {
        this.user_key = user_key;
        this.loan_id = loan_id;
        this.loan_line = loan_line;
        this.loan_avail_line = loan_avail_line;
        this.loan_rate = loan_rate;
        this.apply_period = apply_period;
        this.apply_amount = apply_amount;
        this.apply_time = apply_time;
        this.loan_period = loan_period;
        this.uid = uid;
    }

    public static SingleOutputStreamOperator<String> getFlatMapStream(StreamTableEnvironment tableEnv, StreamExecutionEnvironment env) {
        Table tableRes = tableEnv.sqlQuery(QUERYSQL);
        DataStream<Tuple2<Boolean, TableFdmHfqLoanCenterRiskRecord>> inputStream =
                tableEnv.toRetractStream(tableRes, TableFdmHfqLoanCenterRiskRecord.class);
        SingleOutputStreamOperator<String> out = inputStream.filter(new FilterFunction<Tuple2<Boolean, TableFdmHfqLoanCenterRiskRecord>>() {
            @Override
            public boolean filter(Tuple2<Boolean, TableFdmHfqLoanCenterRiskRecord> value) throws Exception {
                return value.f0;
            }
        }).uid("TableHfqLoanCenterRiskRecord-filter")
        .flatMap(new RichFlatMapFunction<Tuple2<Boolean, TableFdmHfqLoanCenterRiskRecord>, String>() {
            SimpleDateFormat sdf = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                this.sdf = new SimpleDateFormat("yyyy-MM-dd");
            }

            @Override
            public void flatMap(Tuple2<Boolean, TableFdmHfqLoanCenterRiskRecord> value, Collector<String> out) throws Exception {
                try {
                    TableFdmHfqLoanCenterRiskRecord input = value.f1;
                    long apply_time = input.getApply_time().getTime();
                    long currentTimeMillis = System.currentTimeMillis();
                    Integer dateDiff = CommonTimeTools.dateDiff(currentTimeMillis, apply_time, sdf);
                    if (dateDiff <= DATERANGE) {
                        String loan_period = input.getLoan_period();
                        JSONArray loan_periods = JSONArray.parseArray(loan_period);
                        int size = loan_periods.size();
                        int max_loan_period = 0;
                        for (int i = 0; i < size; i++) {
                            JSONObject element = loan_periods.getJSONObject(i);
                            int period = element.getIntValue("period");
                            max_loan_period = (max_loan_period > period) ? max_loan_period : period;
                        }
                        out.collect(consResultJson(input, (max_loan_period == 0) ? null : Integer.valueOf(max_loan_period)));
                    }
                } catch (Exception e) {
                    LOG.error("LoanCenterRiskRecord flatmap excep: " + e.getMessage());
                    LOG.error("LoanCenterRiskRecord flatmap excep input: " + value);
                }
            }

            private String consResultJson(TableFdmHfqLoanCenterRiskRecord record, Integer max_loan_period) {
                JSONObject resJson = new JSONObject();
                resJson.put("user_key", record.getUser_key());
                resJson.put("loan_id", record.getLoan_id() + "");
                resJson.put("loan_line", record.getLoan_line());
                resJson.put("loan_avail_line", record.getLoan_avail_line());
                resJson.put("loan_rate", record.getLoan_rate());
                resJson.put("apply_period", record.getApply_period());
                resJson.put("apply_amount", record.getApply_amount());
                resJson.put("apply_time", Long.valueOf(record.getApply_time().getTime() / 1000L));
                resJson.put("loan_period", max_loan_period);
                resJson.put("uid", record.getUid());
                return resJson.toString();
            }
        });
        return out;
    }

    public Long getUid() {
        return this.uid;
    }

    public String getUser_key() {
        return this.user_key;
    }

    public Long getLoan_id() {
        return this.loan_id;
    }

    public BigDecimal getLoan_line() {
        return this.loan_line;
    }

    public Timestamp getApply_time() {
        return this.apply_time;
    }

    public BigDecimal getLoan_avail_line() {
        return this.loan_avail_line;
    }

    public BigDecimal getLoan_rate() {
        return this.loan_rate;
    }

    public int getApply_period() {
        return this.apply_period;
    }

    public BigDecimal getApply_amount() {
        return this.apply_amount;
    }

    public String getLoan_period() {
        return this.loan_period;
    }
}
